/*
 * @Descripttion: 
 * @version: 
 * @Author: jiuguo.qin
 * @Date: 2020-01-02 10:34:27
 * @LastEditors: jiuguo.qin
 * @LastEditTime: 2020-05-17 15:03:36
 */
#include <dlfcn.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>  
#include <errno.h>

// #include <fcntl.h>  
#include <iostream>
#include <sstream>
#include <fstream>
#include <vector>
#include <sys/un.h>
#include <signal.h>
#include <unistd.h>
#include <errno.h>

#include <netinet/in.h>
#include <netinet/tcp.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include <sys/stat.h>
#include <sys/file.h>
#include <sys/ioctl.h>
#include "server.h"
#include "wrk_logger.h"
#include "wrk_core.h"
#include "ssl.h"
#include "Urlcode.h"
#include "wrk_worker.h"
#include "ae.h"
#include "wrk_setproctitle.h"
#include "wrk_process.h"
#include "wrk_channel.h"
#include "net.h"



#define LISTEN_BACKLOG      10
#define MAX_SETSIZE         1024
#define WRK_HEAD_MAGIC      2020

using namespace std;

extern wrk_worker_process_t *wrk_worker_process;

server::server()
{	
}

server::~server()
{
};

string server::push_host = "";
string server::push_service = "";
string server::push_uri = "";
wrk::WrkLogger* server::logger = NULL;
int server::lock_file_fd = 0;
int server::master_pid = 0;

void server::init_daemon()
{
    pid_t pid;
    
    if ((pid = fork()) != 0)
        exit(0);
    
    setpgid(0, 0);
    
    signal(SIGINT,  SIG_IGN);
    signal(SIGHUP,  SIG_IGN);
    signal(SIGQUIT, SIG_IGN);
    signal(SIGPIPE, SIG_IGN);
    signal(SIGTTOU, SIG_IGN);
    signal(SIGTTIN, SIG_IGN);
    signal(SIGCHLD, SIG_IGN);
    
    if ((pid = fork()) != 0)
        exit(0);    
//   chdir("/");
   close(0);
   umask(0);

   LOG_INFO("init_daemon sucess, running as daemon...");
}


int server::init_tcp_sock(struct sockaddr_in &sin, socklen_t &sin_len, uint32_t port,string ip="")
{
    int fd;   
    bzero(&sin,sizeof(sin));
    sin.sin_family=AF_INET;   
    sin.sin_port=htons(port);
	
	if(ip.empty()){
		sin.sin_addr.s_addr = htonl(INADDR_ANY);
	}else{
		sin.sin_addr.s_addr = inet_addr(ip.c_str());
	}
	
    sin_len=sizeof(sin);

    if((fd=socket(AF_INET,SOCK_STREAM,0))==-1){
        CDEBUG << "create socket error:" << strerror(errno) <<std::endl;
        exit(0); 
    };

    return fd;
}


size_t server::sock_readable(int fd) {
    int n, rc;
    rc = ioctl(fd, FIONREAD, &n);
    return rc == -1 ? 0 : n;
}

size_t server::sock_can_readable(int fd) {
    int n, rc;
    rc = ioctl(fd, FIONREAD, &n);
    return rc == -1 ? 0 : n;
}



wrk_cmd_handler_t server::cmd_handlers[] = {
    { 1, "start", server::master_start_task_handler},
    { 2, "stop", server::master_stop_task_handler},
    { 3, "stats", server::master_stats_task_handler},
    { 4, "upload", server::master_upload_file_handler},
    { 0, NULL, NULL}
};

/**
 * @name: 
 * @test: 
 * @msg: 
 * 这里是入口，客户端的数据从这里过来，魔数不匹配时丢弃请求，后续可考虑TSL
 * master只读取头部 sizeof(wrk_packet_head_t)
 * @param {type} 
 * @return: 
 */
void server::read_cb(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask)
{   
    ssize_t n = 0;

    wrk_packet_head_t *wrk_h = new wrk_packet_head_t();

    ssize_t len = read(fd, wrk_h, sizeof(wrk_packet_head_t));
    aeDeleteFileEvent(eventLoop, fd, AE_READABLE);

    if(len<=0) {
        if(len<0) LOG_WARNING("ignore client request, because read from client error: %s", strerror(errno));
        close(fd);
		return ;
	}

    uint16_t magic = ntohs(wrk_h->magic);
    if(magic != WRK_HEAD_MAGIC) {
        LOG_WARNING("request header magic(%d) is unmatch, ignore this request", magic);
        close(fd);
        return ;
    }

    wrk_cmd_handler_t *p = cmd_handlers;

    uint32_t total_length = ntohl(wrk_h->length);
    uint32_t taskid       = ntohl(wrk_h->taskid);

    wrk_channel_t ch;
    ch.taskid  = taskid;
    ch.command = wrk_h->command;
    ch.fd      = fd;
    ch.length  = total_length;
    memcpy(ch.name, wrk_h->name, sizeof(ch.name));

    for(; p->name!=NULL; p++) {
        if(p->cmd == wrk_h->command) {
            LOG_DEBUG("cmd name: %s", p->name);
            break;
        }
    }

    if(p->name ==NULL) {
        LOG_WARNING("unkown cmd!");
        return ;
    }

    p->handler(fd, ch, eventLoop);
}

void server::response2client(int fd, stringstream &resp, ssize_t n)
{
    int ret = wrk_write(fd, resp.str().c_str(), resp.str().length(), &n);
    if(ret == socket_status_t::ERROR) {
        LOG_ERROR("write response to client error: %s", strerror(errno));
    }
    
    LOG_DEBUG("response:\n%s", resp.str().c_str());
    close(fd);
}

void server::master_start_task_handler(int fd, wrk_channel_t ch, void *data)
{
    stringstream resp;
    ssize_t n = 0;
    
    wrk_worker_process_t *process_ptr = get_one_idle_worker();
    if (process_ptr==NULL) {
        resp << "{\"success\":false, \"message\":\"have no idle worker process\"}\n";
        goto RESP;
    }

    // upload file 不设置状态
    if(ch.command == wrk_cmd_word_t::start) process_ptr->status = wrk_worker_process_t::RUNNING;
    process_ptr->taskid = ch.taskid;

    n = wrk_write_channel(process_ptr->channel[0], &ch, sizeof(wrk_channel_t));

    if(n==WRK_ERROR) 
    {
        switch (errno)
        {
        case EINTR:
            resp << "worker is too busy! you can retry later: " << strerror(errno);
            break;
        default:    
            LOG_ERROR("will kill worker. master(fd:%d) write channel(channelid=%d, ch.fd=%d) into worker(pid:%d) fail, error: %s",
                    server::master_pid, process_ptr->channel[0], ch.fd, process_ptr->pid, strerror(errno));
            kill(process_ptr->pid, SIGKILL);
            resp << strerror(errno);
            break;
        }
        goto RESP;
    }

    LOG_DEBUG("New Taskid:%ld,worker(%d) task start running...", ch.taskid, process_ptr->pid);
    close(fd);
    return;

RESP:
    response2client(fd, resp, n);
}

void server::master_stop_task_handler(int fd, wrk_channel_t ch, void *data)
{
    server::master_stats_task_handler(fd, ch, data);
}

/**
 * @name: 
 * @test: 
 * @msg: 将文件保存起来，path /tmp/dist-wrk-data/taskid/
 * @param {type} 
 * @return: 
 */
void server::master_upload_file_handler(int fd, wrk_channel_t ch, void *data)
{
    server::master_start_task_handler(fd, ch, data);
}


void server::master_stats_task_handler(int fd, wrk_channel_t ch, void *data)
{
    wrk_worker_process_t *process_ptr = get_process_by_taskid(ch.taskid);
    stringstream resp;
    ssize_t n = 0;

    process_ptr = wrk_worker_process;

    if(process_ptr==NULL) {
        LOG_DEBUG("have no found process for taskid: %ld", ch.taskid);
        resp << "{\"success\":false, \"message\":\"have no found process for taskid:"<< ch.taskid << "\"}\n";
        goto RESP;
    }

    n = wrk_write_channel(process_ptr->channel[0], &ch, sizeof(wrk_channel_t));

    if(n==WRK_ERROR) 
    {
        switch (errno)
        {
        case EINTR:
            resp << "worker is too busy! you can retry later: " << strerror(errno);
            break;
        default:    
            LOG_ERROR("will kill worker. master(fd:%d) write channel(channelid=%d, ch.fd=%d) into worker(pid:%d) fail, error: %s",
                    server::master_pid, process_ptr->channel[0], ch.fd, process_ptr->pid, strerror(errno));
            kill(process_ptr->pid, SIGKILL);
            resp << strerror(errno);
            break;
        }
        goto RESP;
    }

    close(fd);
    return;

    RESP:
        response2client(fd, resp, n);
    
}

void server::connected_cb(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask)
{
    socklen_t sin_len;    
    struct sockaddr_in sin;

    int client_fd = accept(fd, (struct sockaddr*)&sin, &sin_len); 
    int  flags = fcntl(client_fd, F_GETFL, 0);
    fcntl(client_fd, F_SETFL, flags | O_NONBLOCK);

    flags = 1;
    setsockopt(client_fd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(flags));
    aeCreateFileEvent(eventLoop, client_fd, AE_READABLE, read_cb, clientData);
}

aeEventLoop* server::ae = NULL;

void server::init_tcp_server()
{
    uint16_t port = cfg_ptr->get_int_value("server", "port");
    if(port<=1024) throw std::string("server port must grater than 1024!");

    long long id;

	socklen_t sin_len;    
    struct sockaddr_in sin;
    
    bind_fd = init_tcp_sock(sin,sin_len, port);

    LOG_DEBUG("init tcp socket success, bind fd=%d", bind_fd);

	int opt = 1;
	setsockopt(bind_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt));
	
    if(bind(bind_fd,(struct sockaddr *)&sin,sizeof(sin)) !=0 ){
        CDEBUG << "bind failed, "<< strerror(errno)<< endl;
        exit(-1);
    };

    if(listen(bind_fd, LISTEN_BACKLOG) !=0) {
        CDEBUG << "listen port:"<<port <<" failed, "<< strerror(errno)<< endl;
        exit(-1);
    }

    /* 添加文件IO事件 */
    int rc = aeCreateFileEvent(ae, bind_fd, AE_READABLE, connected_cb, NULL);
    if(rc != AE_OK) {
        CDEBUG<<"exit(-1), aeCreateFileEvent failed, " << strerror(errno);
        exit(-1);
    }

    LOG_INFO("add bind fd=%d to ae, start tcp server(listen port:%d) success, master pid=%d", bind_fd, port, getpid());
}

int server::bind_fd = -1;

int server::start_tcp_server()
{
    
    aeMain(ae);

    aeDeleteEventLoop(ae);

    return 0;
}

int server::wrk_read(int fd, stringstream &ss, ssize_t *readn)
{
    char buf[MAX_RECV] = {0};
    ssize_t r = 0;
    do {
        ssize_t r = read(fd, buf, sizeof(buf));
        if (r==-1 ) {
            switch (errno) {
                case EAGAIN: return socket_status_t::RETRY;
                default:     return socket_status_t::ERROR;
            }
        }

        ss.write(buf, r);
        *readn += r;
    }while(r == MAX_RECV  && sock_can_readable(fd) > 0);

    return socket_status_t::OK;
}


int server::wrk_write(int fd, const char *buf, ssize_t total_length, ssize_t *writen)
{   
    ssize_t r;
    if ((r = write(fd, buf, total_length)) == -1) {
        switch (errno) {
            case EAGAIN: return socket_status_t::RETRY;
            default:     return socket_status_t::ERROR;
        }
    }
    *writen = (size_t) r;
    return socket_status_t::OK;
}

ssize_t server::push_message(const char *host, const char *service, const char *push_uri, const string &message)
{
    
    struct addrinfo *addrs;
    struct addrinfo hints;
    hints.ai_family = AF_UNSPEC;
    hints.ai_socktype = SOCK_STREAM;

    int rc, index = 1;

    if ((rc = getaddrinfo(host, service, &hints, &addrs)) != 0) {
        const char *msg = gai_strerror(rc);
        LOG_WARNING("ignore push message, unable to resolve(%s:%s)", host, service);
        return -1;
    }

    int fd = socket(addrs->ai_family, addrs->ai_socktype, addrs->ai_protocol);
    int flags = fcntl(fd, F_GETFL, 0);
    fcntl(fd, F_SETFL, flags | O_NONBLOCK);
    if (connect(fd, addrs->ai_addr, addrs->ai_addrlen) == -1) {
        if (errno != EINPROGRESS) {
            LOG_WARNING("ignore push message, connect error(errno=%d), %s", errno, strerror(errno));
            return -1;
        }
    }
    
    flags = 1;
    if(setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(flags)) ==-1){
        LOG_WARNING("ignore push message, setsockopt error(errno=%d), %s", errno, strerror(errno));
        close(fd);
        return -1;
    };

    stringstream ss;
    string str = UrlEncode(message);
    ss << "POST "<<push_uri << " HTTP/1.1\r\n"
       << "Host: "<<host <<"\r\n"
       << "User-Agent: "<< "dist-wrk" << "\r\n"
       << "Accept: */*" << "\r\n"
       << "Content-Type: "<< "application/x-www-form-urlencoded" << "\r\n"
       << "Content-Length:" << str.length() << "\r\n"
       << "\r\n"
       << str;

    ssize_t n =0;

    LOG_DEBUG("push message to http://%s:%s%s \n%s",host, service,push_uri, message.c_str());
    wrk_write(fd, (char *)ss.str().c_str(), ss.str().length(), &n);

    close(fd);

    return n;

}

void server::read_from_worker(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask)
{
    wrk_worker_process_t *p = (wrk_worker_process_t *)clientData;
    ssize_t n;
    stringstream ss;
    int ret = server::wrk_read(fd, ss, &n);

    if(ret == ERROR) {
        LOG_ERROR("read socketpair from worker error(%s), will kill worker(pid=%d)", strerror(errno), p->pid);
        aeDeleteFileEvent(eventLoop, fd, AE_READABLE);
        close(fd);
        kill(9, p->pid);
    }

    if (n==0) {
        return ;
    }

    stringstream push_url;
    // push_url << server::push_uri  << "?id="<<p->taskid;
    
    push_url << server::push_uri << "?id=ch1";
    // LOG_DEBUG("push url:%s", push_url.str().c_str());

    // push_message(push_host.c_str(), push_service.c_str(), push_url.str().c_str(), ss.str());
}

void server::start_worker_process() 
{
    wrk_worker_process = new wrk_worker_process_t[MAX_WORKER_PROCESS_NUM]();
    for(int i=0; i<MAX_WORKER_PROCESS_NUM; i++) {
        pid_t pid = wrk_spawn_process((void *) (intptr_t) i, "worker process");
        aeCreateFileEvent(ae,  wrk_worker_process[i].channel[0], AE_READABLE, read_from_worker, &wrk_worker_process[i]);
    }
}

pid_t server::wrk_spawn_process(void *data, char *name)
{
    intptr_t s = (intptr_t)data;
    if (socketpair(AF_UNIX, SOCK_STREAM, 0, wrk_worker_process[s].channel) == -1) {
        stringstream ss;
        ss << "socketpair error: " << strerror(errno) <<", return !";
        LOG_ERROR(ss.str().c_str());
        throw ss.str();
    }

    int  flags = fcntl(wrk_worker_process[s].channel[0], F_GETFL, 0);
    fcntl(wrk_worker_process[s].channel[0], F_SETFL, flags | O_NONBLOCK);
    fcntl(wrk_worker_process[s].channel[1], F_SETFL, flags | O_NONBLOCK);
    fcntl(wrk_worker_process[s].channel[0], F_SETFL, flags &~ O_NONBLOCK);
    fcntl(wrk_worker_process[s].channel[1], F_SETFL, flags &~ O_NONBLOCK);
    LOG_INFO("set socket(%d, %d) NOBLOCK success", wrk_worker_process[s].channel[0], wrk_worker_process[s].channel[1]);

    flags = 1;
    setsockopt(wrk_worker_process[s].channel[0], IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(flags));
    setsockopt(wrk_worker_process[s].channel[1], IPPROTO_TCP, TCP_NODELAY, &flags, sizeof(flags));
    LOG_INFO("set socket(%d, %d) TCP_NODELAY success",wrk_worker_process[s].channel[0], wrk_worker_process[s].channel[1]);

    pid_t pid = fork();
    if (pid>0) {

        LOG_INFO("fork children process, children pid=%d", pid);

        close(wrk_worker_process[s].channel[1]);
        wrk_worker_process[s].pid = pid;
        wrk_worker_process[s].status = wrk_worker_process_t::IDEL;
        wrk_worker_process[s].name = name;
    }else if(pid==0) {
        wrk_setproctitle(name);

        for(int i=0; i<=s; i++) {
            close(wrk_worker_process[i].channel[0]);
        }
        
        // close(0);
        close(server::bind_fd);
        close(server::lock_file_fd);

        aeStop(server::ae);
        aeDeleteEventLoop(server::ae);

        if(setvbuf(stdout, NULL, _IONBF, 0)) {
            LOG_ERROR("setvbuf error: %s", strerror(errno));
        }else {
            // dup2(wrk_worker_process[s].channel[1], STDOUT_FILENO);
            // dup2(wrk_worker_process[s].channel[1], STDERR_FILENO);
            LOG_INFO("worker(slot=%d) dup2 %d -> %d, %d closed", s, STDOUT_FILENO, wrk_worker_process[s].channel[1], STDOUT_FILENO);
            LOG_INFO("worker(slot=%d) dup2 %d -> %d, %d closed", s, STDERR_FILENO, wrk_worker_process[s].channel[1], STDERR_FILENO);

            WrkWorker *w = new WrkWorker();
            w->run(wrk_worker_process[s].channel[1]);
        }
    }else {
        stringstream ss;
        ss << "fork error: "<< strerror(errno) <<endl;
        LOG_ERROR(ss.str().c_str());
        throw ss.str();
    }

    return pid;
}

int server::init_aeEventLoop()
{

    server::ae = aeCreateEventLoop(MAX_SETSIZE);
    if (!server::ae)
    {
        LOG_ERROR("aeCreateEventLoop failure!");
        throw std::string("aeCreateEventLoop failure!");
    }

    return 0;
}


WrkConfigLoader* server::cfg_ptr = NULL;

void server::init_conf(const string &conf)
{
    stringstream ss;

    server::cfg_ptr = new WrkConfigLoader(conf);

    push_host    = cfg_ptr->get_string_value("push", "host");
    push_service = cfg_ptr->get_string_value("push", "service");
    push_uri     = cfg_ptr->get_string_value("push", "uri");

    MAX_WORKER_PROCESS_NUM = cfg_ptr->get_int_value("server", "wroker_num");

    if(MAX_WORKER_PROCESS_NUM<=0 || MAX_WORKER_PROCESS_NUM>=48) {
        ss << "MAX_WORKER_PROCESS_NUM = "<< MAX_WORKER_PROCESS_NUM <<", invalid, exit" <<endl;
        LOG_ERROR(ss.str().c_str());
        throw std::string(ss.str());
    }
}

void server::init_single_instance()
{
    std::stringstream ss;
    const std::string pid_file = cfg_ptr->get_string_value("server", "pid_file");
    server::lock_file_fd = open(pid_file.c_str(), O_RDWR|O_CREAT, S_IRUSR|S_IWUSR|S_IRGRP|S_IROTH);
    
    if(server::lock_file_fd<0) {
        ss <<"can't open "<<LOCKFILE <<", error:"<<strerror(errno) << std::endl;
        LOG_ERROR(ss.str().c_str());
        throw std::string(ss.str());
    }

    if(flock(server::lock_file_fd, LOCK_NB|LOCK_EX)) {
        ss << "another dist-wrk instance is running!" << std::endl;
        LOG_ERROR(ss.str().c_str());
        throw std::string(ss.str());
    }
    
    int daemon = cfg_ptr->get_int_value("server", "daemon");
    if(daemon) init_daemon();

    ftruncate(server::lock_file_fd, 0);
    ss << getpid();
    write(server::lock_file_fd, ss.str().c_str(), ss.str().length());
    LOG_INFO("save pid:%s into %s success", ss.str().c_str(), pid_file.c_str());

    master_pid = getpid();
}

void server::init_logger()
{
    const std::string logfile = cfg_ptr->get_string_value("server", "log_file");
    const std::string loglevel = cfg_ptr->get_string_value("server", "log_level");

    typedef struct {
        const std::string name;
        wrk::wrk_log_level_t lev;
    }log_level_t;

    log_level_t level[] = {
        {"ERROR", wrk::wrk_log_level_t::ERROR},
        {"WARNING", wrk::wrk_log_level_t::WARNING},
        {"INFO", wrk::wrk_log_level_t::INFO},
        {"DEBUG", wrk::wrk_log_level_t::DEBUG}
    };

    server::logger = wrk::WrkLogger::get_instance();
    server::logger->init_logger(logfile);

    CDEBUG << "init log: "<< logfile << " success! "<<endl;
    // wrk::WrkLogger *logger_ptr = wrk::WrkLogger::get_instance();
    // logger_ptr->init_logger("m.log");

    log_level_t *p = NULL;
    for(int i=0; i<sizeof(level)/sizeof(level[0]); i++) {
        if(level[i].name == loglevel) {
            p = level + i;
            break;
        }
    }

    if(p==NULL) {
        LOG_ERROR("%s", "log level set error! ERROR|WARNING|INFO|DEBUG");
        throw std::string("log level set error! ERROR|WARNING|INFO|DEBUG");
    }

    server::logger->set_log_level(p->lev);

    LOG_INFO("init logger success, logFile = %s, logLevel = %s", logfile.c_str(), loglevel.c_str());
}

void server::init_signal()
{
    wrk_init_signals();
}

int server::run(const string &conf)
{
     try {

        init_conf(conf);

        init_logger();

        init_single_instance();

        init_aeEventLoop();

        init_signal();

        wrk_init_setproctitle();
        
        init_tcp_server();

        wrk_setproctitle("master process");

        start_worker_process();

        start_tcp_server();

        return 0;

    }
    catch(std::string &msg) {
        CDEBUG << msg <<std::endl;
        // LOG_ERROR("%s", msg.c_str());
    }
}
